package com.example.kafka.springboot.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MyConsumer {

//    @KafkaListener(topics = "test",groupId = "default-group")
//    public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
//        String value = record.value();
//        System.out.println(value);
//        System.out.println(record);
//        //⼿动提交offset
//        ack.acknowledge();
//    }

    @KafkaListener(groupId = "default-group",  topicPartitions = {
//            @TopicPartition(topic = "test", partitions = { "0", "1" }),
            @TopicPartition(topic = "test", partitions = "0",
                    partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "2"))
    }, concurrency = "3") //concurrency就是同组下的消费者个数，就是并发消费数，建议⼩于等于分区总数
    public void listenGroup(ConsumerRecord < String, String > record,
                            Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //⼿动提交offset
        ack.acknowledge();
    }

}
